#include "in.h"

int app_CreateStore( struct MQTool *tool )
{
	struct MQSdkContent	*sdk = NULL ;
	
	int			nret = 0 ;
	
	if( tool->store_name == NULL || STRCMP(tool->store_name,==,"") )
	{
		printf( "ERROR : need command parameter '--store_name (store_name)'\n" );
		return -1;
	}
	
	sdk = MQAllocContent( tool->entry.ip , tool->entry.port ) ;
	if( sdk == NULL )
	{
		printf( "MQAllocContent failed\n" );
		return -1;
	}
	else
	{
		printf( "MQAllocContent ok\n" );
	}
	
	nret = MQCreateStore( sdk , tool->store_name ) ;
	if( nret )
	{
		printf( "MQCreateStore failed[%d] , store_name[%s]\n" , nret , tool->store_name );
		MQFreeContent( sdk );
		return -1;
	}
	else
	{
		printf( "MQCreateStore ok\n" );
	}
	
	MQFreeContent( sdk );
	
	return 0;
}

int app_DestroyStore( struct MQTool *tool )
{
	struct MQSdkContent	*sdk = NULL ;
	
	int			nret = 0 ;
	
	if( tool->store_name == NULL || STRCMP(tool->store_name,==,"") )
	{
		printf( "ERROR : need command parameter '--store_name (store_name)'\n" );
		return -1;
	}
	
	sdk = MQAllocContent( tool->entry.ip , tool->entry.port ) ;
	if( sdk == NULL )
	{
		printf( "MQAllocContent failed\n" );
		return -1;
	}
	else
	{
		printf( "MQAllocContent ok\n" );
	}
	
	nret = MQDestroyStore( sdk , tool->store_name ) ;
	if( nret )
	{
		printf( "MQDestroyStore failed[%d] , store_name[%s]\n" , nret , tool->store_name );
		MQFreeContent( sdk );
		return -1;
	}
	else
	{
		printf( "MQCreateStore ok\n" );
	}
	
	MQFreeContent( sdk );
	
	return 0;
}

int app_CreateTopic( struct MQTool *tool )
{
	struct MQSdkContent	*sdk = NULL ;
	
	int			nret = 0 ;
	
	if( tool->store_name == NULL || STRCMP(tool->store_name,==,"") )
	{
		printf( "ERROR : need command parameter '--store_name (store_name)'\n" );
		return -1;
	}
	
	if( tool->topic_name == NULL || STRCMP(tool->topic_name,==,"") )
	{
		printf( "ERROR : need command parameter '--topic_name (topic_name)'\n" );
		return -1;
	}
	
	sdk = MQAllocContent( tool->entry.ip , tool->entry.port ) ;
	if( sdk == NULL )
	{
		printf( "MQAllocContent failed\n" );
		return -1;
	}
	else
	{
		printf( "MQAllocContent ok\n" );
	}
	
	nret = MQCreateTopic( sdk , tool->store_name , tool->topic_name ) ;
	if( nret )
	{
		printf( "MQCreateTopic failed[%d] , store_name[%s] topic_name[%s]\n" , nret , tool->store_name , tool->topic_name );
		MQFreeContent( sdk );
		return -1;
	}
	else
	{
		printf( "MQCreateTopic ok\n" );
	}
	
	MQFreeContent( sdk );
	
	return 0;
}

int app_DestroyTopic( struct MQTool *tool )
{
	struct MQSdkContent	*sdk = NULL ;
	
	int			nret = 0 ;
	
	if( tool->store_name == NULL || STRCMP(tool->store_name,==,"") )
	{
		printf( "ERROR : need command parameter '--store_name (store_name)'\n" );
		return -1;
	}
	
	if( tool->topic_name == NULL || STRCMP(tool->topic_name,==,"") )
	{
		printf( "ERROR : need command parameter '--topic_name (topic_name)'\n" );
		return -1;
	}
	
	sdk = MQAllocContent( tool->entry.ip , tool->entry.port ) ;
	if( sdk == NULL )
	{
		printf( "MQAllocContent failed\n" );
		return -1;
	}
	else
	{
		printf( "MQAllocContent ok\n" );
	}
	
	nret = MQDestroyTopic( sdk , tool->store_name , tool->topic_name ) ;
	if( nret )
	{
		printf( "MQDestroyTopic failed[%d] , store_name[%s] topic_name[%s]\n" , nret , tool->store_name , tool->topic_name );
		MQFreeContent( sdk );
		return -1;
	}
	else
	{
		printf( "MQCreateTopic ok\n" );
	}
	
	MQFreeContent( sdk );
	
	return 0;
}

int app_ProduceMessage( struct MQTool *tool )
{
	struct MQSdkContent	*sdk = NULL ;
	
	int			nret = 0 ;
	
	if( tool->store_name == NULL || STRCMP(tool->store_name,==,"") )
	{
		printf( "ERROR : need command parameter '--store_name (store_name)'\n" );
		return -1;
	}
	
	if( tool->topic_name == NULL || STRCMP(tool->topic_name,==,"") )
	{
		printf( "ERROR : need command parameter '--topic_name (topic_name)'\n" );
		return -1;
	}
	
	sdk = MQAllocContent( tool->entry.ip , tool->entry.port ) ;
	if( sdk == NULL )
	{
		printf( "MQAllocContent failed\n" );
		return -1;
	}
	else
	{
		printf( "MQAllocContent ok\n" );
	}
	
	if( tool->tags == NULL )
		tool->tags = "" ;
	
	if( tool->msg_data == NULL )
		tool->msg_data = "" ;
	
	nret = MQProduceMessage( sdk , tool->store_name , tool->topic_name , tool->tags , tool->msg_data , & (tool->msg_offset) ) ;
	if( nret )
	{
		printf( "MQProduceMessage failed[%d] , store_name[%s] topic_name[%s] tags[%s] msg_data[%s]\n" , nret , tool->store_name , tool->topic_name , tool->tags , tool->msg_data );
		MQFreeContent( sdk );
		return -1;
	}
	else
	{
		printf( "MQProduceMessage ok , store_name[%s] topic_name[%s] tags[%s] msg_data[%s] - msg_offset[%"PRIu64"]\n" , tool->store_name , tool->topic_name , tool->tags , tool->msg_data , tool->msg_offset );
	}
	
	MQFreeContent( sdk );
	
	return 0;
}

int app_ConsumeMessage( struct MQTool *tool )
{
	struct MQSdkContent	*sdk = NULL ;
	
	int			nret = 0 ;
	
	if( tool->store_name == NULL || STRCMP(tool->store_name,==,"") )
	{
		printf( "ERROR : need command parameter '--store_name (store_name)'\n" );
		return -1;
	}
	
	if( tool->topic_name == NULL || STRCMP(tool->topic_name,==,"") )
	{
		printf( "ERROR : need command parameter '--topic_name (topic_name)'\n" );
		return -1;
	}
	
	sdk = MQAllocContent( tool->entry.ip , tool->entry.port ) ;
	if( sdk == NULL )
	{
		printf( "MQAllocContent failed\n" );
		return -1;
	}
	else
	{
		printf( "MQAllocContent ok\n" );
	}
	
	nret = MQBeginSubscribeMessage( sdk , tool->store_name , tool->topic_name , tool->subscribe_group_name ) ;
	if( nret )
	{
		printf( "MQBeginSubscribeMessage failed[%d] , store_name[%s] topic_name[%s] subscribe_group_name[%s]\n" , nret , tool->store_name , tool->topic_name , tool->subscribe_group_name );
		MQFreeContent( sdk );
		return -1;
	}
	else
	{
		printf( "MQBeginSubscribeMessage ok , store_name[%s] topic_name[%s] subscribe_group_name[%s]\n" , tool->store_name , tool->topic_name , tool->subscribe_group_name );
	}
	
	for( ; ; )
	{
		printf( "MQConsumeMessage ...\n" );
		nret = MQConsumeMessage( sdk , & (tool->topic_name) , & (tool->tags) , & (tool->msg_data_len) , & (tool->msg_data) , & (tool->topic_offset) ) ;
		if( nret )
		{
			printf( "MQConsumeMessage failed[%d]\n" , nret );
			MQEndSubscribeMessage( sdk );
			MQFreeContent( sdk );
			return -1;
		}
		else
		{
			printf( "MQConsumeMessage ok , store_name[%s] topic_name[%s] tags[%s] msg_data[%.*s] - topic_offset[%"PRIu64"]\n" , tool->store_name , tool->topic_name , tool->tags , tool->msg_data_len,tool->msg_data , tool->topic_offset );
		}
	}
	
	printf( "MQEndSubscribeMessage\n" );
	MQEndSubscribeMessage( sdk );
	
	MQFreeContent( sdk );
	
	return 0;
}

static void *ProduceMessagesThreadEntry( void *p )
{
	struct BenchmarkStat	*bmstat = (struct BenchmarkStat *) p ;
	struct MQTool		*tool = (struct MQTool *)(bmstat->user_data) ;
	
	struct MQSdkContent	*sdk = NULL ;
	size_t			batch_index ;
	char			msg_data[ 256 ] ;
	uint64_t		span ;
	
	int			nret = 0 ;
	
	MQSetLogEnv( tool->entry.log_level , tool->entry.log_pathfilename );
	
	sdk = MQAllocContent( tool->entry.ip , tool->entry.port ) ;
	if( sdk == NULL )
	{
		printf( "thread[%zu][%ld] - MQAllocContent failed\n" , bmstat->tidx,bmstat->tid );
		return NULL;
	}
	
	if( tool->tags == NULL )
		tool->tags = "" ;
	
	if( tool->msg_data_prefix == NULL )
		tool->msg_data_prefix = "" ;
	
	memset( msg_data , 0x00 , sizeof(msg_data) );
	if( tool->batch_count <= 10 )
		span = 1 ;
	else
		span = tool->batch_count / 10 ;
	
	BEGIN_BENCHMARK( (*bmstat) )
	
	nret = MQConnect( sdk ) ;
	if( nret )
	{
		printf( "thread[%zu][%ld] - MQConnect failed[%d]\n" , bmstat->tidx,bmstat->tid , nret );
		return NULL;
	}
	
	for( batch_index = 0 ; batch_index < tool->batch_count ; batch_index++ )
	{
		snprintf( msg_data , sizeof(msg_data)-1 , "%s-%zu-%zu" , tool->msg_data_prefix , bmstat->tidx+1 , batch_index+1 );
		nret = MQProduceMessage( sdk , tool->store_name , tool->topic_name , tool->tags , msg_data , & (tool->msg_offset) ) ;
		if( nret )
		{
			printf( "thread[%zu][%ld] - MQProduceMessage failed[%d] , store_name[%s] topic_name[%s] tags[%s] msg_data[%s]\n" , bmstat->tidx,bmstat->tid , nret , tool->store_name , tool->topic_name , tool->tags , msg_data );
			MQFreeContent( sdk );
			return NULL;
		}
		else
		{
			bmstat->count++;
			
			if( batch_index % span == 0 )
			{
				printf( "thread[%zu][%ld] - MQProduceMessage ok , store_name[%s] topic_name[%s] tags[%s] msg_data[%s] - msg_offset[%"PRIu64"]\n" , bmstat->tidx,bmstat->tid , tool->store_name , tool->topic_name , tool->tags , msg_data , tool->msg_offset );
			}
		}
	}
	
	MQDisconnect( sdk );
	
	END_BENCHMARK( (*bmstat) )
	DIFF_BENCHMARK( (*bmstat) )
	printf( "thread[%zu][%ld] - MQProduceMessage all ok , count[%"PRIu64"] elapse[%ld.%06ld]s\n" , bmstat->tidx,bmstat->tid , bmstat->count , bmstat->diff_elapse.tv_sec,bmstat->diff_elapse.tv_usec );
	
	MQFreeContent( sdk );
	
	return NULL;
}

int app_ProduceMessages( struct MQTool *tool )
{
	struct BenchmarkStat	*bmstats = NULL ;
	size_t			process_index ;
	
	uint64_t		total_count ;
	double			total_elapse ;
	double			qps ;
	double			min_delay ;
	double			avg_delay ;
	double			max_delay ;
	
	int			nret = 0 ;
	
	if( tool->store_name == NULL || STRCMP(tool->store_name,==,"") )
	{
		printf( "ERROR : need command parameter '--store_name (store_name)'\n" );
		return -1;
	}
	
	if( tool->topic_name == NULL || STRCMP(tool->topic_name,==,"") )
	{
		printf( "ERROR : need command parameter '--topic_name (topic_name)'\n" );
		return -1;
	}
	
	if( tool->process_count == 0 )
		tool->process_count = 1 ;
	
	bmstats = (struct BenchmarkStat *)calloc( sizeof(struct BenchmarkStat) , tool->process_count ) ;
	if( bmstats == NULL )
	{
		printf( "ERROR : calloc failed , errno[%d]\n" , errno );
		return -1;
	}
	memset( bmstats , 0x00 , sizeof(struct BenchmarkStat)*tool->process_count );
	
	for( process_index = 0 ; process_index < tool->process_count ; process_index++ )
	{
		bmstats[process_index].user_data = tool ;
		bmstats[process_index].tidx = process_index ;
		nret = pthread_create( & (bmstats[process_index].tid) , NULL , ProduceMessagesThreadEntry , bmstats+process_index ) ;
		if( nret == -1 )
		{
			printf( "ERROR : pthread_create index[%zu] failed , errno[%d]\n" , process_index , errno );
			free( bmstats );
			return -1;
		}
		else
		{
			printf( "pthread_create index[%zu] ok , tid[%ld]\n" , process_index , bmstats[process_index].tid );
		}
	}
	
	for( process_index = 0 ; process_index < tool->process_count ; process_index++ )
	{
		nret = pthread_join( bmstats[process_index].tid , NULL ) ;
		if( nret == -1 )
		{
			printf( "ERROR : pthread_join index[%zu] tid[%ld] failed , errno[%d]\n" , process_index , bmstats[process_index].tid , errno );
			free( bmstats );
			return -1;
		}
		else
		{
			printf( "pthread_join index[%zu] tid[%ld] ok\n" , process_index , bmstats[process_index].tid );
		}
	}
	
	for( process_index = 0 ; process_index < tool->process_count ; process_index++ )
	{
		printf( "index[%zu] tid[%ld] - count[%"PRIu64"] elapse[%ld.%06ld] - qps[%.6lf] delay[%.6lf]\n" , process_index,bmstats[process_index].tid , bmstats[process_index].count , bmstats[process_index].diff_elapse.tv_sec,bmstats[process_index].diff_elapse.tv_usec , bmstats[process_index].qps , bmstats[process_index].delay );
	}
	
	TotalBenchmarkStat( bmstats , tool->process_count , & total_count , & total_elapse , & qps , & min_delay , & avg_delay , & max_delay );
	free( bmstats );
	
	printf( "--------- PRODUCE BATCH MESSAGES BENCHMARK REPORT ---------\n" );
	printf( " total count[%"PRIu64"]\n" , total_count );
	printf( "total elapse[%.6lf]\n" , total_elapse );
	printf( "         qps[%.6lf]\n" , qps );
	printf( "   min delay[%.6lf]\n" , min_delay );
	printf( "   avg delay[%.6lf]\n" , avg_delay );
	printf( "   max delay[%.6lf]\n" , max_delay );
	
	return 0;
}

static void *ConsumeMessagesThreadEntry( void *p )
{
	struct BenchmarkStat	*bmstat = (struct BenchmarkStat *) p ;
	struct MQTool		*tool = (struct MQTool *)(bmstat->user_data) ;
	
	struct MQSdkContent	*sdk = NULL ;
	size_t			batch_index ;
	uint64_t		span ;
	
	int			nret = 0 ;
	
	MQSetLogEnv( tool->entry.log_level , tool->entry.log_pathfilename );
	
	sdk = MQAllocContent( tool->entry.ip , tool->entry.port ) ;
	if( sdk == NULL )
	{
		printf( "thread[%zu][%ld] - MQAllocContent failed\n" , bmstat->tidx,bmstat->tid );
		return NULL;
	}
	
	nret = MQBeginSubscribeMessage( sdk , tool->store_name , tool->topic_name , tool->subscribe_group_name ) ;
	if( nret )
	{
		printf( "thread[%zu][%ld] - MQBeginSubscribeMessage failed[%d] , store_name[%s] topic_name[%s] subscribe_group_name[%s]\n" , bmstat->tidx,bmstat->tid , nret , tool->store_name , tool->topic_name , tool->subscribe_group_name );
		MQFreeContent( sdk );
		return NULL;
	}
	else
	{
		printf( "thread[%zu][%ld] - MQBeginSubscribeMessage ok , store_name[%s] topic_name[%s] subscribe_group_name[%s]\n" , bmstat->tidx,bmstat->tid , tool->store_name , tool->topic_name , tool->subscribe_group_name );
	}
	
	if( tool->batch_count <= 10 )
		span = 1 ;
	else
		span = tool->batch_count / 10 ;
	
	for( batch_index = 0 ; batch_index < tool->batch_count ; batch_index++ )
	{
		nret = MQConsumeMessage( sdk , & (tool->topic_name) , & (tool->tags) , & (tool->msg_data_len) , & (tool->msg_data) , & (tool->topic_offset) ) ;
		if( nret )
		{
			printf( "thread[%zu][%ld] - MQConsumeMessage failed[%d]\n" , bmstat->tidx,bmstat->tid , nret );
			MQEndSubscribeMessage( sdk );
			MQFreeContent( sdk );
			return NULL;
		}
		else
		{
			if( batch_index == 0 )
			{
				BEGIN_BENCHMARK( (*bmstat) )
			}
			
			bmstat->count++;
			
			if( batch_index % span == 0 )
			{
				printf( "thread[%zu][%ld] - MQConsumeMessage ok , store_name[%s] topic_name[%s] tags[%s] msg_data[%.*s] - topic_offset[%"PRIu64"]\n" , bmstat->tidx,bmstat->tid , tool->store_name , tool->topic_name , tool->tags , tool->msg_data_len,tool->msg_data , tool->topic_offset );
			}
		}
	}
	
	printf( "thread[%zu][%ld] - MQEndSubscribeMessage\n" , bmstat->tidx,bmstat->tid );
	MQEndSubscribeMessage( sdk );
	
	END_BENCHMARK( (*bmstat) )
	DIFF_BENCHMARK( (*bmstat) )
	printf( "thread[%zu][%ld] - MQConsumeMessage all ok , count[%"PRIu64"] elapse[%ld.%06ld]s\n" , bmstat->tidx,bmstat->tid , bmstat->count , bmstat->diff_elapse.tv_sec,bmstat->diff_elapse.tv_usec );
	
	MQFreeContent( sdk );
	
	return NULL;
}

int app_ConsumeMessages( struct MQTool *tool )
{
	struct BenchmarkStat	*bmstats = NULL ;
	size_t			process_index ;
	
	uint64_t		total_count ;
	double			total_elapse ;
	double			qps ;
	double			min_delay ;
	double			avg_delay ;
	double			max_delay ;
	
	int			nret = 0 ;
	
	if( tool->store_name == NULL || STRCMP(tool->store_name,==,"") )
	{
		printf( "ERROR : need command parameter '--store_name (store_name)'\n" );
		return -1;
	}
	
	if( tool->topic_name == NULL || STRCMP(tool->topic_name,==,"") )
	{
		printf( "ERROR : need command parameter '--topic_name (topic_name)'\n" );
		return -1;
	}
	
	if( tool->process_count == 0 )
		tool->process_count = 1 ;
	
	bmstats = (struct BenchmarkStat *)calloc( sizeof(struct BenchmarkStat) , tool->process_count ) ;
	if( bmstats == NULL )
	{
		printf( "ERROR : calloc failed , errno[%d]\n" , errno );
		return -1;
	}
	memset( bmstats , 0x00 , sizeof(struct BenchmarkStat)*tool->process_count );
	
	for( process_index = 0 ; process_index < tool->process_count ; process_index++ )
	{
		bmstats[process_index].user_data = tool ;
		bmstats[process_index].tidx = process_index ;
		nret = pthread_create( & (bmstats[process_index].tid) , NULL , ConsumeMessagesThreadEntry , bmstats+process_index ) ;
		if( nret == -1 )
		{
			printf( "ERROR : pthread_create index[%zu] failed , errno[%d]\n" , process_index , errno );
			free( bmstats );
			return -1;
		}
		else
		{
			printf( "pthread_create index[%zu] ok , tid[%ld]\n" , process_index , bmstats[process_index].tid );
		}
	}
	
	for( process_index = 0 ; process_index < tool->process_count ; process_index++ )
	{
		nret = pthread_join( bmstats[process_index].tid , NULL ) ;
		if( nret == -1 )
		{
			printf( "ERROR : pthread_join index[%zu] tid[%ld] failed , errno[%d]\n" , process_index , bmstats[process_index].tid , errno );
			free( bmstats );
			return -1;
		}
		else
		{
			printf( "pthread_join index[%zu] tid[%ld] ok\n" , process_index , bmstats[process_index].tid );
		}
	}
	
	for( process_index = 0 ; process_index < tool->process_count ; process_index++ )
	{
		printf( "index[%zu] tid[%ld] - count[%"PRIu64"] elapse[%ld.%06ld] - qps[%.6lf] delay[%.6lf]\n" , process_index,bmstats[process_index].tid , bmstats[process_index].count , bmstats[process_index].diff_elapse.tv_sec,bmstats[process_index].diff_elapse.tv_usec , bmstats[process_index].qps , bmstats[process_index].delay );
	}
	
	TotalBenchmarkStat( bmstats , tool->process_count , & total_count , & total_elapse , & qps , & min_delay , & avg_delay , & max_delay );
	free( bmstats );
	
	printf( "--------- CONSUME BATCH MESSAGES BENCHMARK REPORT ---------\n" );
	printf( " total count[%"PRIu64"]\n" , total_count );
	printf( "total elapse[%.6lf]\n" , total_elapse );
	printf( "         qps[%.6lf]\n" , qps );
	printf( "   min delay[%.6lf]\n" , min_delay );
	printf( "   avg delay[%.6lf]\n" , avg_delay );
	printf( "   max delay[%.6lf]\n" , max_delay );
	
	return 0;
}

int app_ProduceDelayMessage( struct MQTool *tool )
{
	struct MQSdkContent	*sdk = NULL ;
	
	int			nret = 0 ;
	
	if( tool->store_name == NULL || STRCMP(tool->store_name,==,"") )
	{
		printf( "ERROR : need command parameter '--store-name (store_name)'\n" );
		return -1;
	}
	
	if( tool->topic_name == NULL || STRCMP(tool->topic_name,==,"") )
	{
		printf( "ERROR : need command parameter '--topic-name (topic_name)'\n" );
		return -1;
	}
	
	if( tool->delay_seconds <= 0 )
	{
		printf( "ERROR : need command parameter '--delay-seconds (seconds)'\n" );
		return -1;
	}
	
	sdk = MQAllocContent( tool->entry.ip , tool->entry.port ) ;
	if( sdk == NULL )
	{
		printf( "MQAllocContent failed\n" );
		return -1;
	}
	else
	{
		printf( "MQAllocContent ok\n" );
	}
	
	if( tool->tags == NULL )
		tool->tags = "" ;
	
	if( tool->msg_data == NULL )
		tool->msg_data = "" ;
	
	nret = MQProduceDelayMessage( sdk , tool->delay_seconds , tool->store_name , tool->topic_name , tool->tags , tool->msg_data , & (tool->delay_msg_offset) ) ;
	if( nret )
	{
		printf( "MQProduceMessage failed[%d] , delay_seconds[%"PRIu32"] store_name[%s] topic_name[%s] tags[%s] msg_data[%s]\n" , nret , tool->delay_seconds , tool->store_name , tool->topic_name , tool->tags , tool->msg_data );
		MQFreeContent( sdk );
		return -1;
	}
	else
	{
		printf( "MQProduceMessage ok , delay_seconds[%"PRIu32"] store_name[%s] topic_name[%s] tags[%s] msg_data[%s] - delay_msg_offset[%"PRIu64"]\n" , tool->delay_seconds , tool->store_name , tool->topic_name , tool->tags , tool->msg_data , tool->delay_msg_offset );
	}
	
	MQFreeContent( sdk );
	
	return 0;
}

int app_CancelDelayMessage( struct MQTool *tool )
{
	struct MQSdkContent	*sdk = NULL ;
	
	int			nret = 0 ;
	
	if( tool->store_name == NULL || STRCMP(tool->store_name,==,"") )
	{
		printf( "ERROR : need command parameter '--store-name (store_name)'\n" );
		return -1;
	}
	
	sdk = MQAllocContent( tool->entry.ip , tool->entry.port ) ;
	if( sdk == NULL )
	{
		printf( "MQAllocContent failed\n" );
		return -1;
	}
	else
	{
		printf( "MQAllocContent ok\n" );
	}
	
	nret = MQCancelDelayMessage( sdk , tool->store_name , tool->delay_msg_offset ) ;
	if( nret )
	{
		printf( "MQCancelDelayMessage failed[%d] , store_name[%s] delay_msg_offset[%"PRIu64"]\n" , nret , tool->store_name , tool->delay_msg_offset );
		MQFreeContent( sdk );
		return -1;
	}
	else
	{
		printf( "MQCancelDelayMessage ok , store_name[%s] delay_msg_offset[%"PRIu64"]\n" , tool->store_name , tool->delay_msg_offset );
	}
	
	MQFreeContent( sdk );
	
	return 0;
}

int app_ParseMessageFile( struct MQTool *tool )
{
	struct MQFileMapping	*msg_file = NULL ;
	
	uint64_t		read_len ;
	uint64_t		msg_offset ;
	uint8_t			topic_name_len ;
	char			*topic_name = NULL ;
	uint16_t		tags_len ;
	char			*tags = NULL ;
	uint32_t		msg_data_len ;
	char			*msg_data = NULL ;
	
	int			nret = 0 ;
	
	nret = LoadMessageDataFile( NULL , NULL , NULL , 0 , tool->file , & msg_file , NULL ) ;
	if( nret )
	{
		printf( "load msg file '%s' failed[%d]\n" , tool->file , nret );
		return -1;
	}
	
	if( msg_file == NULL )
	{
		printf( "msg file '%s' is cleaned\n" , tool->file );
		return -1;
	}
	
	for( ; ; )
	{
		read_len = MemRead_MessageDataUnit( msg_file->fill_ptr , & msg_offset , & topic_name_len , & topic_name , & tags_len , & tags , & msg_data_len , & msg_data ) ;
		if( read_len == 0 )
		{
			printf( "not end of file\n" );
			break;
		}
		else if( read_len == UINT64_MAX )
		{
			printf( "end of file\n" );
			break;
		}
		else
		{
			printf( "msg_offset[%"PRIu64"] topic_name[%.*s] tags[%.*s] msg_data[%.*s]\n" , msg_offset , topic_name_len,topic_name , tags_len,tags , msg_data_len,msg_data );
		}
		
		msg_file->fill_ptr += read_len ;
	}
	
	munmap( msg_file->file_mmap , msg_file->file_size );
	free( msg_file );
	
	return 0;
}

int app_ParseTopicFile( struct MQTool *tool )
{
	struct MQFileMapping	*topic_file = NULL ;
	uint64_t		topic_offset ;
	
	uint64_t		read_len ;
	uint64_t		msg_offset ;
	
	int			nret = 0 ;
	
	nret = LoadTopicDataFile( NULL , NULL , NULL , NULL , 0 , tool->file , & topic_file , NULL ) ;
	if( nret )
	{
		printf( "load topic file '%s' failed[%d]\n" , tool->file , nret );
		return -1;
	}
	
	if( topic_file == NULL )
	{
		printf( "topic file '%s' is cleaned\n" , tool->file );
		return -1;
	}
	
	for( ; ; )
	{
		read_len = MemRead_TopicDataUnit( topic_file->fill_ptr , & topic_offset , & msg_offset ) ;
		if( read_len == 0 )
		{
			printf( "not end of file\n" );
			break;
		}
		else if( read_len == UINT64_MAX )
		{
			printf( "end of file\n" );
			break;
		}
		else
		{
			printf( "topic_offset[%"PRIu64"] msg_offset[%"PRIu64"]\n" , topic_offset , msg_offset );
		}
		
		topic_file->fill_ptr += read_len ;
	}
	
	munmap( topic_file->file_mmap , topic_file->file_size );
	free( topic_file );
	
	return 0;
}

int app_ParseDelayMessageFile( struct MQTool *tool )
{
	struct MQFileMapping	*delay_msg_file = NULL ;
	uint64_t		delay_msg_offset ;
	
	uint64_t		read_len ;
	uint64_t		delay_msg_id ;
	uint32_t		delay_seconds ;
	char			is_sent ;
	uint8_t			store_name_len ;
	char			*store_name = NULL ;
	uint8_t			topic_name_len ;
	char			*topic_name = NULL ;
	uint16_t		tags_len ;
	char			*tags = NULL ;
	uint32_t		msg_data_len ;
	char			*msg_data = NULL ;
	
	int			nret = 0 ;
	
	nret = LoadDelayMessageDataFile( NULL , NULL , NULL , 0 , tool->file , & delay_msg_file , NULL ) ;
	if( nret )
	{
		printf( "load delay msg file '%s' failed[%d]\n" , tool->file , nret );
		return -1;
	}
	
	if( delay_msg_file == NULL )
	{
		printf( "delay msg file '%s' is cleaned\n" , tool->file );
		return -1;
	}
	
	delay_msg_offset = delay_msg_file->begin_offset + MQ_DELAY_MSG_FILE_HEAD_SIZE ;
	for( ; ; )
	{
		read_len = MemRead_DelayMessageDataUnit( delay_msg_file->fill_ptr , & delay_msg_id , & delay_seconds , & is_sent , & store_name_len , & store_name , & topic_name_len , & topic_name , & tags_len , & tags , & msg_data_len , & msg_data ) ;
		if( read_len == 0 )
		{
			printf( "not end of file\n" );
			break;
		}
		else if( read_len == UINT64_MAX )
		{
			printf( "end of file\n" );
			break;
		}
		else
		{
			printf( "delay_msg_offset[%"PRIu64"] delay_msg_id[%"PRIu64"] delay_seconds[%"PRIu32"] is_sent[%c] store_name[%.*s] topic_name[%.*s] tags[%.*s] msg_data[%.*s]\n" , delay_msg_offset , delay_msg_id , delay_seconds , is_sent , store_name_len,store_name , topic_name_len,topic_name , tags_len,tags , msg_data_len,msg_data );
		}
		
		delay_msg_file->fill_ptr += read_len ;
		delay_msg_offset += read_len ;
	}
	
	munmap( delay_msg_file->file_mmap , delay_msg_file->file_size );
	free( delay_msg_file );
	
	return 0;
}

